package com.fanxuankai.ksql.function.udf;

import io.confluent.ksql.function.udf.Udf;
import io.confluent.ksql.function.udf.UdfDescription;
import io.confluent.ksql.function.udf.UdfParameter;

import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;

/**
 * @author fanxuankai
 */
@UdfDescription(name = "IF_ELSE", description = "If Else", author = "fanxuankai")
public class IfElse {
    private static final String P_CONDITION = "condition";
    private static final String P_VAL_IF_TRUE = "valIfTrue";
    private static final String P_VAL_IF_FALSE = "valIfFalse";

    @Udf(description = "If else with Integer parameters.")
    public Integer ifElse(@UdfParameter(P_CONDITION) Boolean condition, @UdfParameter(P_VAL_IF_TRUE) Integer valIfTrue,
                          @UdfParameter(P_VAL_IF_FALSE) Integer valIfFalse) {
        return eval(condition, valIfTrue, valIfFalse);
    }

    @Udf(description = "If else with Long parameters.")
    public Long ifElse(@UdfParameter(P_CONDITION) Boolean condition, @UdfParameter(P_VAL_IF_TRUE) Long valIfTrue,
                       @UdfParameter(P_VAL_IF_FALSE) Long valIfFalse) {
        return eval(condition, valIfTrue, valIfFalse);
    }

    @Udf(description = "If else with Double parameters.")
    public Double ifElse(@UdfParameter(P_CONDITION) Boolean condition, @UdfParameter(P_VAL_IF_TRUE) Double valIfTrue,
                         @UdfParameter(P_VAL_IF_FALSE) Double valIfFalse) {
        return eval(condition, valIfTrue, valIfFalse);
    }

    @Udf(description = "If else with String parameters.")
    public String ifElse(@UdfParameter(P_CONDITION) Boolean condition, @UdfParameter(P_VAL_IF_TRUE) String valIfTrue,
                         @UdfParameter(P_VAL_IF_FALSE) String valIfFalse) {
        return eval(condition, valIfTrue, valIfFalse);
    }

    @Udf(description = "If else with Boolean parameters.")
    public Boolean ifElse(@UdfParameter(P_CONDITION) Boolean condition, @UdfParameter(P_VAL_IF_TRUE) Boolean valIfTrue,
                          @UdfParameter(P_VAL_IF_FALSE) Boolean valIfFalse) {
        return eval(condition, valIfTrue, valIfFalse);
    }

    @Udf(description = "If else with BigDecimal parameters.")
    public BigDecimal ifElse(@UdfParameter(P_CONDITION) Boolean condition,
                             @UdfParameter(P_VAL_IF_TRUE) BigDecimal valIfTrue,
                             @UdfParameter(P_VAL_IF_FALSE) BigDecimal valIfFalse) {
        return eval(condition, valIfTrue, valIfFalse);
    }

    @Udf(description = "If else with Time parameters.")
    public Time ifElse(@UdfParameter(P_CONDITION) Boolean condition, @UdfParameter(P_VAL_IF_TRUE) Time valIfTrue,
                       @UdfParameter(P_VAL_IF_FALSE) Time valIfFalse) {
        return eval(condition, valIfTrue, valIfFalse);
    }

    @Udf(description = "If else with Date parameters.")
    public Date ifElse(@UdfParameter(P_CONDITION) Boolean condition, @UdfParameter(P_VAL_IF_TRUE) Date valIfTrue,
                       @UdfParameter(P_VAL_IF_FALSE) Date valIfFalse) {
        return eval(condition, valIfTrue, valIfFalse);
    }

    @Udf(description = "If else with Timestamp parameters.")
    public Timestamp ifElse(@UdfParameter(P_CONDITION) Boolean condition,
                            @UdfParameter(P_VAL_IF_TRUE) Timestamp valIfTrue,
                            @UdfParameter(P_VAL_IF_FALSE) Timestamp valIfFalse) {
        return eval(condition, valIfTrue, valIfFalse);
    }

    @Udf(description = "If else with ByteBuffer parameters.")
    public ByteBuffer ifElse(@UdfParameter(P_CONDITION) Boolean condition,
                             @UdfParameter(P_VAL_IF_TRUE) ByteBuffer valIfTrue,
                             @UdfParameter(P_VAL_IF_FALSE) ByteBuffer valIfFalse) {
        return eval(condition, valIfTrue, valIfFalse);
    }

    private <T> T eval(Boolean condition, T valIfTrue, T valIfFalse) {
        return (condition != null && condition) ? valIfTrue : valIfFalse;
    }
}
